package tailfile

import (
	"context"
	"fmt"

	"github.com/Shopify/sarama"
	"github.com/hpcloud/tail"
	"github.com/sirupsen/logrus"

	"logproject/kafka"
	"logproject/model"
)

type Tailf struct {
	tailObj  *tail.Tail
	EtcdConf []model.CollerctEntry
	Changes  bool
	cancel   context.CancelFunc
	ctx      context.Context
}
type TailFileRegister map[string]*Tailf

func newTailf() *Tailf {
	ctx, cancel := context.WithCancel(context.Background())
	return &Tailf{
		cancel: cancel,
		ctx:    ctx,
	}
}

func Init(collectEntryList []model.CollerctEntry, kafkaobj *kafka.KafkaObj) (registerTasks TailFileRegister, tailfstructForSave *Tailf, err error) {

	//按allFile中的文件路径 创建多个文件监控
	config := tail.Config{
		Location:  &tail.SeekInfo{Offset: 0, Whence: 2}, //打开后从哪来开始写 2表示从末位
		ReOpen:    true,                                 //自动打开
		Follow:    true,                                 //文件大小超过后 切割 自动跟随新的文件
		MustExist: false,
		Poll:      true,
	}
	registerTasks = make(TailFileRegister)
	tailfstructForSave = newTailf()
	for _, filepath := range collectEntryList {
		tailfstruct := newTailf()
		//tailfstructForSave.EtcdConf = append(tailfstruct.EtcdConf, filepath)
		tailfstruct.tailObj, err = tail.TailFile(filepath.Path, config)
		registerTasks[filepath.Topic] = tailfstruct
		fmt.Printf("%p\n", tailfstruct.tailObj)
		if err != nil {
			logrus.Errorf("TailFile failed path:%s\n", filepath.Path)
			continue
		}

		go tailfstruct.Run(filepath.Topic, kafkaobj)

	}

	return
}

func (t *Tailf) Tailfs() <-chan *tail.Line {
	return t.tailObj.Lines
}

func (t *Tailf) TailFileName() string {
	return t.tailObj.Filename
}

//读取日志 发送到kafka
func (t *Tailf) Run(topic string, kafkaobj *kafka.KafkaObj) {

	//循环读出
	for {
		select {
		case <-t.ctx.Done():
			logrus.Infof("%s task stopped", topic)
			return
		case line, ok := <-t.tailObj.Lines:
			if !ok || len(line.Text) == 1 {
				logrus.Warn("tail file err,filename:", t.tailObj.Filename)
				continue
			}

			//把读出的line封装成sarama.ProducerMessage丢到kafka的管道里 用于异步读取
			msg := &sarama.ProducerMessage{
				Topic: topic,
				Value: sarama.StringEncoder(line.Text),
			}
			kafkaobj.ToMsgChan(msg)
		}

	}

}

func ChangeTailTasks(tailfstructForSave *Tailf, registerTasks TailFileRegister, kafkaobj *kafka.KafkaObj, registerpcs kafka.Registerpc) {

	for {
		//检测到变化 对比之前的task开启变化后的协程
		if tailfstructForSave.Changes == true {
			tailfstructForSave.Changes = false
			config := tail.Config{
				Location:  &tail.SeekInfo{Offset: 0, Whence: 2}, //打开后从哪来开始写 2表示从末位
				ReOpen:    true,                                 //自动打开
				Follow:    true,                                 //文件大小超过后 切割 自动跟随新的文件
				MustExist: false,
				Poll:      true,
			}
			newRegisterTask := make(TailFileRegister)
			//fmt.Println("1",tailfile.TailFileRegister)
			if len(tailfstructForSave.EtcdConf) != 0 {
				for _, filepath := range tailfstructForSave.EtcdConf {
					//原来已有任务
					if _, ok := registerTasks[filepath.Topic]; ok {
						newRegisterTask[filepath.Topic] = registerTasks[filepath.Topic]
						continue
					}
					//新任务
					tai := newTailf()
					var err error
					tai.tailObj, err = tail.TailFile(filepath.Path, config)
					newRegisterTask[filepath.Topic] = tai
					//registerTasks = newRegisterTask
					if err != nil {
						logrus.Errorf("TailFile failed path:%s\n", filepath.Path)
						continue
					}
					go tai.Run(filepath.Topic, kafkaobj)
				}
			}
			for key, v := range registerTasks {
				found := false
				for newkey := range newRegisterTask {
					if key == newkey {
						found = true
						break
					}
				}
				if !found {
					v.cancel()
					//registerTasks = newRegisterTask
					err := v.tailObj.Stop()
					if err != nil {
						logrus.Errorf("stop tailfile failed", err)
						return
					}
					logrus.Info("stopped taifile")
				}
			}
			registerTasks = newRegisterTask
			err, res := kafkaobj.NewPartitionConsumer(tailfstructForSave.EtcdConf, registerpcs)
			if err != nil {
				logrus.Error("NewPartitionConsumer failed err=", err)
				continue
			}
			registerpcs = res
		}

	}

}
